/******************************************************************************
* Copyright (C) 2020 ShenZhen Powerdata Information Technology Co.,Ltd
* All Rights Reserved.
* 本软件为***研制。未经本公司正式书面同意，其他任何个人、团体不得使用、
* 复制、修改或发布本软件.
*****************************************************************************/

package com.xxl.job.listener.mq.entity;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

/**
* @Title:
* @ClassName:  消费者
* @Description:TODO  
* @author: xgf
* @date:   2020年2月22日 下午2:57:00   
* @since   JDK1.8
* @history 2020年2月22日 TODO 新建
*/
public class B {

	public static void main(String[] args) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("job_listener_consumer");
		consumer.setNamesrvAddr("172.16.11.12:9876");
		consumer.subscribe("test_quick_topic", "*");
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				MessageExt msg = msgs.get(0);
				
				
				try {
					String topic = msg.getTopic();
					String tags = msg.getTags();
					String keys = msg.getKeys();
					String msgContent = new String (msg.getBody() , RemotingHelper.DEFAULT_CHARSET);
					System.out.println(topic +  "  " + tags +  "  " + keys +  "  " + msgContent +  "  " );
					
					
				} catch (Exception e) {
					String topic = msg.getTopic();
					String tags = msg.getTags();
					String keys = msg.getKeys();
					System.err.println(topic +  "  " + tags +  "  " + keys +  "  " + msg.getReconsumeTimes());
					
					//第三次就回收
					if(msg.getReconsumeTimes() >= 2 ) {
						//加入补偿表中
						System.out.println("加入补偿表");
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;	
					}
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;	
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;	
			}
		});
		
		consumer.start();
		
		
	}
}
